package cs.rsa.ts14.circuitbreaker;

import cs.rsa.ts14.circuitbreaker.exceptions.ArgumentOutOfRangeException;
import cs.rsa.ts14.circuitbreaker.states.CircuitBreakerState;
import cs.rsa.ts14.circuitbreaker.states.Closed;
import cs.rsa.ts14.circuitbreaker.states.HalfOpen;
import cs.rsa.ts14.circuitbreaker.states.Open;
import cs.rsa.ts14dist.common.Constants;
import cs.rsa.ts14dist.cookie.remote.ProtectedRemoteCookieService;
import cs.rsa.ts14dist.cookie.remote.StandardRemoteCookieService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;

public class CircuitBreakerImpl<Type> implements CircuitBreaker {
	Logger logger = LoggerFactory.getLogger(CircuitBreakerImpl.class);
	private ReentrantLock lock = null;
	private CircuitBreakerState state = null;
	private ExecutorService threadExecutor = null;

	private int failures;
	private int threshold;
	private long timeoutNanoSeconds = 0;

	public CircuitBreakerImpl(int threshold, long timeoutNanoSeconds) throws ArgumentOutOfRangeException {

		if (threshold < 1 || timeoutNanoSeconds < 1)
			throw new ArgumentOutOfRangeException("Both threshold and timeout should be greater than 0");

		this.lock = new ReentrantLock();
		this.threshold = threshold;
		this.timeoutNanoSeconds = timeoutNanoSeconds;

		moveToClosedState();
	}

	public boolean isClosed() {
		return state instanceof Closed;
	}

	public boolean isOpen() {
		return state instanceof Open;
	}

	public boolean isHalfOpen() {
		return state instanceof HalfOpen;
	}

	public void moveToClosedState() {
		state = new Closed(this);
	}

	public void moveToOpenState() {
		state = new Open(this);
	}

	public void moveToHalfOpenState() {
		state = new HalfOpen(this);
	}

	public void increaseFailureCount() {
		failures++;
	}
	
	public void prepareFailureCountForHalfOpen() {
		this.failures = threshold - 1;
	}

	public void resetFailureCount() {
		failures = 0;
	}

	public boolean thresholdReached() {
		return failures >= threshold;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public <T> T doAction(Callable callable, Class<T> type) {
		boolean run = true;
		String result = "fault";
		logger.info("state is " + state.getClass());
		while (run && !thresholdReached()) {
			run = false;
			Future<String> futureResult = null;
			try {

				if (lock.tryLock(timeoutNanoSeconds, TimeUnit.NANOSECONDS)) {
					state.protectedCodeIsAboutToBeCalled();

					if (threadExecutor == null || threadExecutor.isTerminated())
						threadExecutor = Executors.newSingleThreadScheduledExecutor();

					futureResult = threadExecutor.submit(callable);
					result = futureResult.get(timeoutNanoSeconds, TimeUnit.NANOSECONDS);

					if (futureResult.isCancelled())
						throw new TimeoutException("Timeout marker reached");
				}

				if (lock.isHeldByCurrentThread())
					state.protectedCodeHasBeenCalled();

			} catch (Exception e) {
				logger.info("Exception, current state is " + state.getClass(), e);
				state.exceptionOccurred(e);
				run = true;
			} finally {
				if (lock.isHeldByCurrentThread())
					lock.unlock();

				if (futureResult != null) 
					futureResult.cancel(false);				
				
				if (threadExecutor != null && !threadExecutor.isShutdown())
					threadExecutor.shutdown();

			}
		}

		return type.cast(result);
	}

	public void doAction(Runnable runnable) {
		Thread thread = null;
		boolean run = true;
		long timeout = 0;

		while (run && !thresholdReached()) {
			run = false;
			try {

				if (lock.tryLock(timeoutNanoSeconds, TimeUnit.NANOSECONDS)) {
					state.protectedCodeIsAboutToBeCalled();

					timeout = System.nanoTime() + timeoutNanoSeconds;

					thread = new Thread(runnable);
					thread.setDaemon(true);
					thread.start();

					while (thread.isAlive()) {
						if (System.nanoTime() < timeout)
							throw new TimeoutException("Timeout marker reached");
					}
				}

				if (lock.isHeldByCurrentThread())
					state.protectedCodeHasBeenCalled();

			} catch (Exception e) {
				state.exceptionOccurred(e);

				if (thread != null && thread.isAlive())
					thread.stop();

				run = true;
			} finally {
				if (lock.isHeldByCurrentThread())
					lock.unlock();
			}
		}
	}

	public void close() {
		if (lock.isHeldByCurrentThread()) {
			lock.unlock();
			moveToClosedState();
		}
	}

	public void open() {
		try {
			if (lock.isHeldByCurrentThread() || lock.tryLock(timeoutNanoSeconds, TimeUnit.NANOSECONDS))
				moveToOpenState();
		} catch (Exception e) {
			System.out.println("Exception during transistion to Open state");
			e.printStackTrace();
		}
	}

	public int getThreshold() {
		return threshold;
	}

	public long getTimeoutNanoSeconds() {
		return timeoutNanoSeconds;
	}

    @SuppressWarnings({ "rawtypes" })
    public static void main(String[] args) throws ArgumentOutOfRangeException, InterruptedException {
        StandardRemoteCookieService cookie = new StandardRemoteCookieService(Constants.DIGITALOCEAN_INSTANCE_IP, Constants.COOKIE_REST_PORT);
        CircuitBreaker cb = new CircuitBreakerImpl(3, 6000 * 1000000);

        ProtectedRemoteCookieService cookieService = new ProtectedRemoteCookieService(cookie, cb);
        for(int i=0; i<10; i++) {
            String fortune = null;
            try {
            	fortune = cookieService.getNextCookie();
                cb.moveToOpenState();
                Thread.sleep(11000);
            } catch (IOException e) {
                fortune = "error";
            }
            System.out.println(fortune);
        }
    }
}